package com.leiyuee.flink.tools.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.joda.time.DateTime;

public class GenericMapDataExample {
    public static void main(String[] args) throws Exception {

        // =========================================== GenericMapData 使用 ==========================================
        /*Map<String, Object> map = new HashMap<>();
        map.put("key1", "value1");
        map.put("key2", "value2");
        map.put("key3", "value3");
        map.put("key4", "value4");

        GenericMapData mapData = new GenericMapData(map);
        Object key1 = mapData.get("key1");
        if (key1 instanceof String)
            System.out.println("String: " + key1);
        if (key1 instanceof StringData)
            System.out.println("StringData: " + key1);
        System.out.println("valueArray: " + mapData.valueArray());*/



        // =========================================== 通过 keyBy 分组，然后聚合数据 ==========================================

        // 场景：通过订单 order_id 将 商品数据字段聚合订单详情，一个订单对应一个详情
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        DataStream<JSONObject> source = environment.fromElements(
                "{\"order_id\":\"123456701\",\"product_id\":\"P001\",\"product_name\":\"Product A\",\"quantity\":2,\"price\":50.00}",
                "{\"order_id\":\"1234567002\",\"product_id\":\"P002\",\"product_name\":\"Product B\",\"quantity\":2,\"price\":150.00}",
                "{\"order_id\":\"123456703\",\"product_id\":\"P003\",\"product_name\":\"Product C\",\"quantity\":2,\"price\":250.00}",
                "{\"order_id\":\"123456701\",\"order_date\":\"2023-11-29\",\"customer_name\":\"John Doe\",\"total_amount\":100.00}",
                "{\"order_id\":\"1234567002\",\"order_date\":\"2023-11-30\",\"customer_name\":\"John Doe2\",\"total_amount\":100.00}",
                "{\"order_id\":\"123456703\",\"order_date\":\"2023-11-31\",\"customer_name\":\"John Doe3\",\"total_amount\":100.00}"
        ).map(JSON::parseObject).returns(JSONObject.class);

        source.keyBy(json -> json.getString("order_id")).reduce((json1, json2) -> {

            if (json1.getString("order_id").equals(json2.getString("order_id"))) {
                System.out.println("\n\n" + new DateTime() + "\t" + "GenericMapDataExample");
                System.out.println("json1: " + json1);
                System.out.println("json2: " + json2);


                // 场景 1：一个订单对应一个详情
                //JSONObject result;
                //result = json1;
                //result.put("customer_name", json2.getString("customer_name"));
                //result.put("order_date", json2.getString("order_date"));
                //result.put("total_amount", json2.getString("total_amount"));


                // 场景 2：一个订单对应多个个详情
                JSONObject result;
                result = json1;
                result.put("item", json2);

                System.out.println("result: " + result);
                return result;
            }
            return null;
        });
        environment.execute("GenericMapDataExample");


    }
}